package com.atguigu.rabbitmq.eight;

import com.atguigu.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

/*
* 消费者
* 这里对死信队列进行演示，C1
* 也就是消息传递给正常交换机，正常队列，如果一切正常，直接传递给C1
* 如果在正常队列那里出现了问题，将消息传递给死信交换机，死信交换机给死信队列，
* 死信队列里面的数据由C2读取
* */
public class Consumer01 {

    /*普通交换机名称*/
    public static final String NORMAL_EXCHANGE="normal_exchange";

    /*死信交换机名称*/
    public static final String DEAD_EXCHANGE="dead_exchange";

    /*普通队列的名称*/
    public static final String NORMAL_QUEUE="normal_queue";

    /*死信队列的名称*/
    public static final String DEAD_QUEUE="dead_queue";


    public static void main(String[] args) throws IOException, TimeoutException {
        /*通过connection创建一个channel*/
        Channel channel = RabbitMqUtils.getChannel();
        /*声明死信交换机和普通交换机，类型为direct*/
        channel.exchangeDeclare(NORMAL_EXCHANGE, "direct");
        channel.exchangeDeclare(DEAD_EXCHANGE, "direct");
        /*声明死信队列*/
        channel.queueDeclare(DEAD_QUEUE, false,false,false,null);
        /*声明普通队列*/
        /*1.队列名称 2.是否持久化 3.是否共享 4.是否自动删除 5.其他参数：设置ttl过期时间*/
        /*当ttl过期时间到达之后，这个普通队列就变成了死信队列*/
        HashMap<String, Object> params = new HashMap<>();
        /*正常队列设置死信交换机,正常队列出现问题，就连接在死信交换机上面*/
        params.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        /*设置死信routingkey , 死信队列和死信交换机连接起来*/
        params.put("x-dead-letter-routing-key", "lisi");
        /*设置正常队列的长度限制*/
//        params.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false,false,false,params);

        /*将普通交换机和普通队列之间进行绑定*/
        channel.queueBind(NORMAL_QUEUE , NORMAL_EXCHANGE, "zhangsan");
        /*将死信交换机和死信队列进行绑定*/
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");
        System.out.println("等待接收消息.....");


        /*接收消息成功的回调函数*/
        DeliverCallback deliverCallback=(consumerTag,message)->{
            String msg=new String(message.getBody(),"UTF-8");
            /*如果消费的消息是info5,*/
            if(msg.equals("info5")){
                System.out.println("Consumer01_接收的消息是："+msg+"，这个消息是被C1拒绝的消息");
                /*拒绝消费消息，并且不放回队列，那么就变成了死信队列*/
                channel.basicReject(message.getEnvelope().getDeliveryTag(), false);
            }else {
                System.out.println("Consumer01_接收的消息是："+msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };

        /*消息接收失败的回调函数*/
        CancelCallback cancelCallback=consumerTag->{
            System.out.println("消息消费失败~");
        };
        /*消费消息*/
        /*第二个参数是否自动确认,false就是手动应答的意思*/
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, cancelCallback);

    }

}















